package cn.monkey.transport.netty.duplex;

import cn.monkey.transport.core.Exchange;
import cn.monkey.transport.core.ExchangeContext;
import cn.monkey.transport.netty.NettyExchange;
import io.netty.util.AttributeKey;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

public class SimpleDuplexExchange implements DuplexExchange {
    protected final NettyExchange nettyDeviceExchange;
    protected final Exchange duplexExchange;
    protected final DuplexInbound duplexDeviceInbound;
    protected final DuplexOutbound duplexDeviceOutbound;

    protected final Scheduler scheduler;

    public SimpleDuplexExchange(NettyExchange nettyDeviceExchange,
                                Exchange duplexExchange,
                                Scheduler scheduler) {
        this.nettyDeviceExchange = nettyDeviceExchange;
        this.duplexExchange = duplexExchange;
        this.duplexDeviceInbound = this.newInbound();
        this.duplexDeviceOutbound = this.newOutbound();
        this.scheduler = scheduler;
    }

    protected DuplexInbound newInbound() {
        return new SimpleDuplexInbound(nettyDeviceExchange.inbound(), duplexExchange.outbound(), this.scheduler);
    }

    protected DuplexOutbound newOutbound() {
        return new SimpleDuplexOutbound(nettyDeviceExchange.outbound(), duplexExchange.inbound(), this.scheduler);
    }

    @Override
    public String id() {
        return this.nettyDeviceExchange.id();
    }

    @Override
    public ExchangeContext context() {
        return this.nettyDeviceExchange.context();
    }

    @Override
    public <T> T setAttribute(String s, T t) {
        return this.nettyDeviceExchange.setAttribute(s, t);
    }

    @Override
    public <T> T getAttribute(String s) {
        return this.nettyDeviceExchange.getAttribute(s);
    }

    @Override
    public DuplexInbound inbound() {
        return this.duplexDeviceInbound;
    }

    @Override
    public DuplexOutbound outbound() {
        return this.duplexDeviceOutbound;
    }

    @Override
    public boolean isActive() {
        return this.nettyDeviceExchange.isActive();
    }

    @Override
    public Mono<Void> close() {
        Mono<Void> m1 = this.inbound().close();
        Mono<Void> m2 = this.outbound().close();
        return Flux.merge(m1, m2).then();
    }

    @Override
    public DuplexExchange bind() {
        AttributeKey<Exchange> attributeKey = AttributeKey.valueOf(Exchange.class, Exchange.KEY);
        this.nettyDeviceExchange.setAttribute(attributeKey.name(), this);
        return this;
    }
}
